package com.atguigu.pro4

import com.atguigu.pro3.OrderEvent
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * @description: individual patterns,个体模式可以包括单例模式和循环模式,单例模式只接收一个事件,而循环模式可以接收多个,可以在一个个体模式后追加两次,也就是指定循环次数
 * @time: 2021/4/6 14:30 订单超时事件
 * @author: baojinlong
 * */


// 定义到账事件样例类

object TxMatchWithJoin {
  def main(args: Array[String]): Unit = {
    /**
     * 匹配出现4次 start.times(4)     匹配出现2,3或者4次,并且尽可能多地重复 start.times(2,4).greedy
     * 匹配出现0或者4次 start.times(4).optional 匹配出现1次或多次start.oneOrMore
     * 匹配出现2,3或者4次 start.times(2,4) 匹配出现0次,2次或多次,并且尽可能多地重复 start.timesOrMore(2).optional.greedy
     * 个体模式的条件,每隔模式都需要指定触发条件,作为模式是否接受事件进入的判断依据
     * cep中的个体模式主要通过调用.where or until来指定条件
     * 严格近邻 strict contiguity 宽松relaxed 非确定性宽松紧邻non-deterministic relaxed contiguity
     * 匹配事件的提取
     */
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 不影响输出顺序
    environment.setParallelism(1)
    // 有事件时间的都设置这个时间
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    // 0.读取订单数据
    val orderEventStream: KeyedStream[OrderEvent, String] = environment.readTextFile("src/main/resources/OrderLog.csv")
      .map(data => {
        val arr: Array[String] = data.split(",")
        OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000)
      .filter(_.eventType == "pay")
      .keyBy(_.txId)

    // 读取到账事件数据
    val receiptEventStream: KeyedStream[ReceiptEvent, String] = environment.readTextFile("src/main/resources/ReceiptLog.csv")
      .map(data => {
        val arr: Array[String] = data.split(",")
        ReceiptEvent(arr(0), arr(1), arr(2).toLong)
      })
      // 升序数据也默认延时1毫秒
      .assignAscendingTimestamps(_.timestamp * 1000)
      .keyBy(_.txId)

    val resultStream: DataStream[(OrderEvent, ReceiptEvent)] = orderEventStream.intervalJoin(receiptEventStream)
      .between(Time.seconds(-3), Time.seconds(5))
      // 不包含
      // .lowerBoundExclusive()
      .process(new TxMatchWithJoinResult)

    resultStream.print("resultStream")


    environment.execute("order timeout job")
  }

}


class TxMatchWithJoinResult extends ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)] {
  override def processElement(left: OrderEvent, right: ReceiptEvent, ctx: ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    // 左边的连接右边的输出即可,但是不能够匹配的却不能输出了
    out.collect((left, right))
  }
}